#ifndef _QUEUE_H_
#define _QUEUE_H_

#include <atomic>
#include <memory>
#include <cassert>
#include <cstring>
#include <unistd.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#include <iostream>

/**
    基于共享内存的通信队列实现
*/

namespace sp{

// 一个通信库，可以支持进程间通信，也可以支持进程内通信，基于共享内存实现，使用内存中的原子锁做互斥，使用数组实现环形队列。
/** 内存空间
 *  front | rear | enqueue_mutex | dequeue_mutex | data 
 **/
template<class T>
class BlockingQueue{
private:
    int capacity;                      // 容量
    int* front;                        // 起始位置
    int* rear;                         // 结束位置
    key_t shm_id;                      // 共享内存标识符
    char* ptr;                         // 共享内存起始地址
    std::atomic<bool>* enqueue_mutex;  // 生产者锁
    std::atomic<bool>* dequeue_mutex;  // 消费者锁
    T* data_ptr;                       // 数据起始地址
    // 加锁与解锁
    void lock(std::atomic<bool>* mutex);
    void unlock(std::atomic<bool>* mutex);
    // 禁用拷贝
    BlockingQueue(const BlockingQueue&) = delete;
    BlockingQueue& operator= (const BlockingQueue&) = delete;
public:
    // 相关函数
    BlockingQueue(int capacity);
    ~BlockingQueue();
    int size();
    bool empty();
    bool full();
    T front_value();   // 获取队首的值，但不取出
    bool dequeue(T& data);                  // 读取数据，非阻塞     
    void wait_dequeue(T& data);             // 读取数据，阻塞
    bool dequeue(T* data, int len);         // 读取多个数据，非阻塞
    void wait_dequeue(T* data, int len);    // 读取多个数据，阻塞
    bool enqueue(T data);                   // 发送数据，非阻塞
    void wait_enqueue(T data);              // 发送数据，阻塞
    bool enqueue(T* data, int len);         // 发送多个数据，非阻塞
    void wait_enqueue(T* data, int len);    // 发送多个数据，阻塞
};

template<class T>
BlockingQueue<T>::BlockingQueue(int capacity){
    this->capacity = capacity;
    int element_size = sizeof(T);
    size_t size = 10 + element_size*capacity;
    shm_id = shmget(IPC_PRIVATE, size, IPC_CREAT | 0666);
    assert(shm_id >= 0);
    ptr = (char*)shmat(shm_id, 0, 0);
    front = new (ptr)int(0);
    rear = new (ptr+4)int(0);
    enqueue_mutex = new (ptr+8)std::atomic<bool>(false);
    dequeue_mutex = new (ptr+9)std::atomic<bool>(false);
    data_ptr = (T*)(ptr + 10);
}

template<class T>
BlockingQueue<T>::~BlockingQueue(){
    shmdt(ptr);
    shmid_ds sds;
    shmctl(shm_id, IPC_RMID, &sds);
}

template<class T>
void BlockingQueue<T>::lock(std::atomic<bool>* mutex){
    bool old_value = false;
    bool new_value = true;
    while(mutex->compare_exchange_strong(old_value, new_value)){
        sleep(0.00001);
    }
}

template<class T>
void BlockingQueue<T>::unlock(std::atomic<bool>* mutex){
    assert(*mutex == true);
    *mutex = false;
}

template<class T>
int BlockingQueue<T>::size(){
    return (*rear+capacity-*front)%capacity;
}

template<class T>
bool BlockingQueue<T>::empty(){
    return *front == *rear;
}

template<class T>
bool BlockingQueue<T>::full(){
    return (*rear+1)%capacity == *front;
}

template<class T>
T BlockingQueue<T>::front_value(){
    assert(!empty());
    return data_ptr[*front];
}

template<class T>
bool BlockingQueue<T>::dequeue(T& data){                 // 读取数据，非阻塞
    lock(dequeue_mutex);
    if(empty()){
        unlock(dequeue_mutex);
        return false;
    }
    data = data_ptr[*front];
    *front = (*front+1)%capacity;
    unlock(dequeue_mutex);
    return true;
}

template<class T>
void BlockingQueue<T>::wait_dequeue(T& data){            // 读取数据，阻塞
    lock(dequeue_mutex);
    while(empty());
    data = data_ptr[*front];
    *front = (*front+1)%capacity;
    unlock(dequeue_mutex);
}

template<class T>
bool BlockingQueue<T>::dequeue(T* data, int len){        // 读取多个数据，非阻塞
    lock(dequeue_mutex);
    if(size() < len){
        unlock(dequeue_mutex);
        return false;
    }
    if(*front + len <= capacity){ // 不分段
        memcpy(data, data_ptr + *front, len);
    }else{ // 分两段
        memcpy(data, data_ptr + *front, capacity-*front);
        memcpy(data+capacity-*front, data_ptr, len-(capacity-*front));
    }
    memcpy(data, data_ptr + *front, len);
    *front = (*front+len)%capacity;
    unlock(dequeue_mutex);
    return true;
}

template<class T>
void BlockingQueue<T>::wait_dequeue(T* data, int len){    // 读取多个数据，阻塞
    lock(dequeue_mutex);
    while(size() < len){ // 发送接收命令可以有延迟，每次不成功 10us后再判断。
        sleep(0.00001);
    }
    if(*front + len <= capacity){ // 不分段
        memcpy(data, data_ptr + *front, len);
    }else{ // 分两段
        memcpy(data, data_ptr + *front, capacity-*front);
        memcpy(data+capacity-*front, data_ptr, len-(capacity-*front));
    }
    *front = (*front+len)%capacity;
    unlock(dequeue_mutex);
}

template<class T>
bool BlockingQueue<T>::enqueue(T data){                  // 发送数据，非阻塞
    lock(enqueue_mutex);
    if(full()){
        unlock(enqueue_mutex);
        return false;
    }
    data_ptr[*rear] = data;
    *rear = (*rear+1)%capacity;
    unlock(enqueue_mutex);
    return true;
}

template<class T>
void BlockingQueue<T>::wait_enqueue(T data){             // 发送数据，阻塞
    lock(enqueue_mutex);
    while(full());
    data_ptr[*rear] = data;
    *rear = (*rear+1)%capacity;
    unlock(enqueue_mutex);
}

template<class T>
bool BlockingQueue<T>::enqueue(T* data, int len){         // 发送多个数据，非阻塞
    lock(enqueue_mutex);
    if(capacity - size() < len){
        unlock(enqueue_mutex);
        return false;
    }
    if(*rear + len <= capacity){ // 不分段
        memcpy(data_ptr + *rear, data, len);
    }else{ // 分两段
        memcpy(data_ptr + *rear, data, capacity-*rear);
        memcpy(0, data+capacity-*rear, len-(capacity-*rear));
    }
    *rear = (*rear+len)%capacity;
    unlock(enqueue_mutex);
    return true;
}

template<class T>
void BlockingQueue<T>::wait_enqueue(T* data, int len){    // 发送多个数据，阻塞
    lock(enqueue_mutex);
    while(capacity - size() < len);
    if(*rear + len <= capacity){ // 不分段
        memcpy(data_ptr + *rear, data, len);
    }else{ // 分两段
        memcpy(data_ptr + *rear, data, capacity-*rear);
        memcpy(data_ptr, data+capacity-*rear, len-(capacity-*rear));
    }
    *rear = (*rear+len)%capacity;
    unlock(enqueue_mutex);
}

};

#endif